package com.gxlevi

import java.util.Properties

import com.alibaba.fastjson.JSON
import com.gxlevi.bean.{ClickLog, ClickLogWide, Message}
import com.gxlevi.task.{ChannelAreaTask, ChannelBrowser, ChannelBrowserTask, ChannelFreshnessTask, ChannelNetWork, ChannelNetWorkTask, ChannelPvUv, ChannelPvUvTask, ChannelRealHot, ChannelRealHotTask, PreprocessTask}
import com.gxlevi.util.GlobalConfigUtil
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.api.{CheckpointingMode, TimeCharacteristic}
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.api.scala._
import org.apache.flink.runtime.state.filesystem.FsStateBackend
import org.apache.flink.streaming.api.environment.CheckpointConfig
import org.apache.flink.streaming.api.functions.{AssignerWithPeriodicWatermarks, AssignerWithPunctuatedWatermarks}
import org.apache.flink.streaming.api.watermark.Watermark
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer010


object App {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    env.setParallelism(1)

    env.enableCheckpointing(5000)
    env.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE)
    env.getCheckpointConfig.setMinPauseBetweenCheckpoints(1000)
    env.getCheckpointConfig.setCheckpointTimeout(60000)
    env.getCheckpointConfig.setMaxConcurrentCheckpoints(1)
    env.getCheckpointConfig.enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION)
    env.setStateBackend(new FsStateBackend("hdfs://node01:8020/flink-checkpoint"))

    val properties = new Properties()
    properties.setProperty("bootstrap.servers", GlobalConfigUtil.bootstrapServers)
    //properties.setProperty("zookeeper.connect", GlobalConfigUtil.zookeeperConnect)
    properties.setProperty("group.id", GlobalConfigUtil.groupId)
    properties.setProperty("enable.auto.commit", GlobalConfigUtil.enableAutoCommit)
    properties.setProperty("auto.commit.interval.ms", GlobalConfigUtil.autoCommitIntervalMs)
    properties.setProperty("auto.offset.reset", GlobalConfigUtil.autoOffsetReset)

    val consumer: FlinkKafkaConsumer010[String] = new FlinkKafkaConsumer010[String](
      GlobalConfigUtil.inputTopic,
      new SimpleStringSchema(),
      properties
    )
    val kafkaDataStream = env.addSource(consumer)

    val messageDataStream = kafkaDataStream.map {
      msgJson =>
        val jsonObject = JSON.parseObject(msgJson)
        val count = jsonObject.getLong("count")
        val timestamp = jsonObject.getLong("timeStamp")
        val message = jsonObject.getString("message")
        Message(ClickLog(message), count, timestamp)
    }

    val waterMarkDataStream: DataStream[Message] = messageDataStream.assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks[Message] {
      var currentTimestamp = 0L
      var maxDelayTime = 2000L

      override def getCurrentWatermark: Watermark = {
        new Watermark(currentTimestamp - maxDelayTime)
      }

      override def extractTimestamp(element: Message, previousElementTimestamp: Long): Long = {
        currentTimestamp = Math.max(element.timeStamp, previousElementTimestamp)
        currentTimestamp
      }
    })

    val clickWideDataStream: DataStream[ClickLogWide] = PreprocessTask.process(waterMarkDataStream)


    //实时频道热点分析:统计频道被访问的数量
    //    ChannelRealHotTask.process(clickWideDataStream)
    //实时频道PV/UV分析:针对频道的PV、UV进行不同时间维度的分析(小时,天,月)
    //    ChannelPvUvTask.process(clickWideDataStream)
    //实时频道用户新鲜度分析:分析网站每小时、每天、每月活跃的新老用户占比
    //    ChannelFreshnessTask.process(clickWideDataStream)
    //实时频道地域分析:查看地域相关的PV/UV、用户新鲜度
    //    ChannelAreaTask.process(clickWideDataStream)
    //实时运营商分析:分析出来中国移动、中国联通、中国电信等运营商的指标。来分析，流量的主要来源是哪个运营商的，这样就可以进行较准确的网络推广
    //    ChannelNetWorkTask.process(clickWideDataStream)
    //实时频道浏览器分析:需要分别统计不同浏览器（或者客户端）的占比(pv,uv,新用户,老用户)
    ChannelBrowserTask.process(clickWideDataStream)
    env.execute("App")

  }
}
